package com.gis507.mapreduce.intertest;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import com.gis507.mapreduce.Utils.UtilsClass;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class AISInterpolationMapper extends Mapper<LongWritable,Text, Text, AisClassifyBean> {

    private Text outK = new Text();
    private AisClassifyBean outV = new AisClassifyBean();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String line = value.toString();

        String[] comment = line.split(",");

        //String mmsi, String shipType, String nacStatusCN, double draught,
        // double heading, double course, double speed, String dest,
        // long unixTime, double lon_d, double lat_d, int seaRange
        String mmsi = comment[0];
        String shipType = comment[1];
        String nacStatusCN = comment[2];
        double draught = UtilsClass.turnToDouble(comment[3]);
        double heading = UtilsClass.turnToDouble(comment[4]);
        double course = UtilsClass.turnToDouble(comment[5]);
        double speed = UtilsClass.turnToDouble(comment[6]);
        String dest = comment[7];
        long unixTime = UtilsClass.turnToLong(comment[8]);
        double lon_d = UtilsClass.turnToDouble(comment[9]);
        double lat_d = UtilsClass.turnToDouble(comment[10]);

        //fengzhuang
        outK.set(mmsi);

        outV.setMmsi(mmsi);
        outV.setShipType(shipType);
        outV.setNacStatusCN(nacStatusCN);
        outV.setDraught(draught);
        outV.setHeading(heading);
        outV.setCourse(course);
        outV.setSpeed(speed);
        outV.setDest(dest);
        outV.setUnixTime(unixTime);
        outV.setLon_d(lon_d);
        outV.setLat_d(lat_d);

        //get the value of seaRange
        int seaRange = UtilsClass.getSeaRangeNumber(outV);

        //set seaRange for outV
        outV.setSeaRange(seaRange);

        //write for Mapper
        context.write(outK,outV);
    }
}
